package second.actor

import scala.actors.{Actor, Future}
import scala.collection.mutable
import scala.collection.mutable.ListBuffer
import scala.io.Source

/**
  * Created by feng on 24/11/2018.
  */


class Task extends Actor {
  override def act(): Unit = {

    loop{
      react{
        case SubmitTask(filename)=>{
          val resultTask = Source.fromFile(filename).getLines().flatMap(_.split(" ")).map((_,1)).toList.groupBy(_._1).mapValues(_.size)
          sender ! ResultTask(resultTask)
        }
        case StopTask=>{
          exit()
        }
      }
    }


  }
}


case class SubmitTask(filename:String)
case class ResultTask(result:Map[String,Int])
case object StopTask

object WordCount {


  def main(args: Array[String]): Unit = {
    val replySet = new mutable.HashSet[Future[Any]]
    val resultList = new ListBuffer[ResultTask]()

    val files = Array[String]("/Users/feng/Desktop/test/wordcount/b.txt","/Users/feng/Desktop/test/wordcount/a.txt")

    for (f<-files){
      val actor = new Task
      val reply = actor.start() !! SubmitTask(f)
      replySet += reply
    }

    while(replySet.size>0){
      val toCompute = replySet.filter(_.isSet)
      for (f <- toCompute){
        val result = f().asInstanceOf[ResultTask]
        resultList += result
        replySet -= f
      }
      Thread.sleep(100)
    }

    val fr = resultList.flatMap(_.result).groupBy(_._1).mapValues(_.foldLeft(0)(_+_._2))
    println(fr)

  }

}
